/*****************************************************************************
 * Copyright (C) 2008 EnterpriseDB Corporation.
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 as published by
 * the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, see http://www.gnu.org/licenses or write to the
 * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 *
 * You can contact EnterpriseDB, Inc. via its website at
 * http://www.enterprisedb.com
 *
 ****************************************************************************/
package com.edb.gridsql.communication;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Level;

import com.edb.gridsql.common.util.Property;
import com.edb.gridsql.common.util.XLogger;
import com.edb.gridsql.communication.message.NodeMessage;
import com.edb.gridsql.exception.XDBServerException;

/**
 * Factory class to create Connectors
 * 
 * @see com.edb.gridsql.communication.AbstractConnector
 * 
 *  
 * @version 1.0
 */
public class ConnectorFactory {

    private static final XLogger logger = XLogger
            .getLogger(ConnectorFactory.class);

    /**
     * Constant to mark Connector undefined
     */
    public static final int CONNECTOR_TYPE_UNKNOWN = -1;

    /**
     * Constant to define local (in-process) Connector
     * 
     * @see com.edb.gridsql.communication.LocalConnector
     */
    public static final int CONNECTOR_TYPE_LOCAL = 0;

    /**
     * Constant to define TCP Socket Connector Removed, channels are used
     * instead
     */
    public static final int CONNECTOR_TYPE_SOCKET = 1;

    /**
     * Constant to define TCP Socket Connector using J2SE 1.4 channels
     */
    public static final int CONNECTOR_TYPE_CHANNEL = 2;

    /**
     * Constant to define UDP Socket (broadcast) Connector
     */
    public static final int CONNECTOR_TYPE_BROADCAST = 3;

    /**
     * msgQTable is used for simple communications between in-process (running
     * within the same Java VM) Nodes. Key of the Hashtable is Node's number,
     * value is a message queue (actually is java.util.LinkedList). Entries are
     * created by the com.edb.gridsql..communication.LocalConnector instances, 
     * once they are initialized.
     */
    private static final ConcurrentHashMap<Integer, BlockingQueue<NodeMessage>> msgQTable = new ConcurrentHashMap<Integer, BlockingQueue<NodeMessage>>();

    private static HashMap<Integer, AbstractConnector> localConnectors = new HashMap<Integer, AbstractConnector>();

    private static HashMap<Integer, AbstractConnector> socketConnectors = new HashMap<Integer, AbstractConnector>();

    private static HashMap<Integer, AbstractConnector> broadcastConnectors = new HashMap<Integer, AbstractConnector>();

    /**
     * Factory method to construct Connector between specified Nodes
     * @param sourceNodeNum 
     * @param targetNodeNum 
     * @throws com.edb.gridsql.exception.XDBServerException 
     * @return 
     */

    public static final AbstractConnector getConnector(int sourceNodeNum,
            int targetNodeNum) throws XDBServerException {
        String connectorTypeParam = "xdb.connector." + sourceNodeNum + "."
                + targetNodeNum;

        // First look for explicit connector type
        int connectorType = Property.getInt(connectorTypeParam,
                CONNECTOR_TYPE_UNKNOWN);

        if (connectorType == CONNECTOR_TYPE_UNKNOWN) {
            try {
                InetAddress sourceAddr = InetAddress.getByName(Property.get(
                        "xdb."
                                + ((sourceNodeNum == 0) ? "coordinator"
                                        : "node." + sourceNodeNum) + ".host",
                        "localhost"));
                InetAddress targetAddr = InetAddress.getByName(Property.get(
                        "xdb."
                                + ((targetNodeNum == 0) ? "coordinator"
                                        : "node." + targetNodeNum) + ".host",
                        "localhost"));

                if (sourceAddr.equals(targetAddr)) {
                    connectorType = CONNECTOR_TYPE_LOCAL;
                } else {
                    connectorType = CONNECTOR_TYPE_CHANNEL;
                }
            } catch (UnknownHostException uhe) {
                // ???
                connectorType = CONNECTOR_TYPE_BROADCAST;
            }
        }

        switch (connectorType) {
        case CONNECTOR_TYPE_LOCAL:
            logger.log(Level.DEBUG,
                    "Requested local Connector from %0% to %1%", new Object[] {
                            new Integer(sourceNodeNum),
                            new Integer(targetNodeNum) });

            // Source node already exist, no need to create it
            // and create source connector before target node to
            // avoid recursion when target node will create connector back
            AbstractConnector sourceConnector = getLocalConnector(sourceNodeNum);

            if (msgQTable.get(targetNodeNum) == null) {
                // Method creates node if it doesn't exist yet
                if (targetNodeNum == 0) {
                    XDBServerException ex = new XDBServerException(
                            "Could not found Coordinator's LocalConnector - configuration inconsistent");
                    logger.throwing(ex);
                    throw ex;
                }
                NodeAgent.getNodeAgent(targetNodeNum);
            }

            return sourceConnector;

        case CONNECTOR_TYPE_SOCKET:
        case CONNECTOR_TYPE_CHANNEL:
            synchronized (socketConnectors) {
                Integer src = new Integer(sourceNodeNum);
                AbstractConnector connector = socketConnectors.get(src);
                if (connector == null) {
                    try {
                        Class socketClass = Class
                                .forName("com.edb.gridsql.communication.SocketConnector");
                        connector = (AbstractConnector) socketClass
                                .getConstructor(new Class[] { int.class })
                                .newInstance(new Object[] { src });
                        socketConnectors.put(src, connector);
                    } catch (Throwable t) {
                        return getLocalConnector(sourceNodeNum);
                    }
                }
                return connector;
            }

        case CONNECTOR_TYPE_BROADCAST:
            return getBroadcastConnector(sourceNodeNum);
        default:
            XDBServerException ex = new XDBServerException(
                    "Requested connector type does not supported");
            logger.throwing(ex);
            throw ex;
        }
    }

    /**
     * Returns connector to local node
     * 
     * @param sourceNodeNum
     * @return
     */
    private static AbstractConnector getLocalConnector(int sourceNodeNum) {
        synchronized (localConnectors) {
            AbstractConnector connector = localConnectors.get(sourceNodeNum);

            if (connector == null) {
                connector = new LocalConnector(sourceNodeNum, msgQTable);
                localConnectors.put(sourceNodeNum, connector);
            }

            return connector;
        }
    }

    /**
     * Returns broadcast connector
     * 
     * @param sourceNodeNum
     * @return a broadcast connector
     */
    public static AbstractConnector getBroadcastConnector(int sourceNodeNum) {
        synchronized (broadcastConnectors) {
            AbstractConnector connector = broadcastConnectors
                    .get(sourceNodeNum);

            if (connector == null) {
                String connectorClassName = Property
                        .get("xdb.broadcast.connector");
                if (connectorClassName == null) {
                    return null;
                } else {
                    try {
                        Class broadcastClass = Class
                                .forName(connectorClassName);
                        connector = (AbstractConnector) broadcastClass
                                .getConstructor(new Class[] { int.class })
                                .newInstance(
                                        new Object[] { new Integer(
                                                sourceNodeNum) });
                        broadcastConnectors.put(new Integer(sourceNodeNum),
                                connector);
                    } catch (Throwable t) {
                        logger.catching(t);
                        return null;
                    }
                }
            }

            return connector;
        }
    }
}